package homework2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.*;
import org.apache.hadoop.hbase.client.*;
import java.io.IOException;

public class functools {
    // 设置对应私有成员
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;

    public static void main(String[] args) throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
        // 主函数目前先什么也不做，但是功能都是实现好的
    }

    public static void init() throws IOException {
        // 设置对应参数，比如dir和admin
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");
        connection = ConnectionFactory.createConnection(configuration);
        admin = connection.getAdmin();
    }

    public static void close() throws IOException {
        // 关闭对应admin链接，在每个功能完成之后都应该关闭对应链接，保证程序安全运行
        admin.close();
    }

    public static void createTable(String tableName, String[] fields) throws IOException {
        init();
        // 创建表并且判断是否存在，如果存在就先删除表，在根据对应field创建对应行
        TableName tablename = TableName.valueOf(tableName);
        if (admin.tableExists(tablename)) {
            System.out.println("table is exists!");
            admin.disableTable(tablename);
            admin.deleteTable(tablename);// 删除原来的表
        }
        TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tablename);
        for (String str : fields) {
            tableDescriptor.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(str)).build());
            admin.createTable(tableDescriptor.build());
        }
    }

    public static void addRecord(String tableName, String row, String[] fields, String[] values) throws IOException {
        init();
        // 添加一条记录
        Table table = connection.getTable(TableName.valueOf(tableName));
        for (int i = 0; i != fields.length; i++) {
            Put put = new Put(row.getBytes());
            String[] cols = fields[i].split(":");
            put.addColumn(cols[0].getBytes(), cols[1].getBytes(), values[i].getBytes());
            table.put(put);
        }
        table.close();
    }

    public static void scanColumn(String tableName, String column) throws IOException {
        init();
        // 遍历对应行，输出
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(column));
        ResultScanner scanner = table.getScanner(scan);
        for (Result result = scanner.next(); result != null; result = scanner.next()) {
            showRecord(result);
        }
        table.close();
        close();
    }

    // 格式化输出
    public static void showRecord(Result result) {
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println("RowName:"
                    + new String(Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())) + " ");
            System.out.println("Timetamp:" + cell.getTimestamp() + " ");
            System.out.println("column Family:"
                    + new String(Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()))
                    + " ");
            System.out.println("row Name:" + new String(
                    Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()))
                    + " ");
            System.out.println("value:"
                    + new String(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()))
                    + " ");
        }
    }

    public static void modifyData(String tableName, String row, String column, String val) throws IOException {
        // 修改表格对应数据
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(row.getBytes());
        put.addColumn(column.getBytes(), null, val.getBytes());
        table.put(put);
        table.close();
        close();
    }

    public static void deleteRow(String tableName, String row) throws IOException {
        // 删除表格对应数据
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(row.getBytes());
        table.delete(delete);
        table.close();
        close();
    }
}
